package com.tinysand.advert.mysql.listener.impl;

import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import com.tinysand.advert.mysql.TemplateHolder;
import com.tinysand.advert.mysql.dto.BinlogRowData;
import com.tinysand.advert.mysql.dto.TableTemplate;
import com.tinysand.advert.mysql.listener.Listener;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.Serializable;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @author TINY
 */
@Slf4j
@Component
public class AggregationListener
        implements BinaryLogClient.EventListener {

    private String databaseName;
    private String tableName;
    private Map<String, Listener> listenerMap = new HashMap<>();

    private final TemplateHolder templateHolder;

    @Autowired
    public AggregationListener(TemplateHolder templateHolder) {
        this.templateHolder = templateHolder;
    }

    private String genKey(String databaseName, String tableName) {
        return databaseName + ":" + tableName;
    }

    public void register(String databaseName, String tableName,
                         Listener listener) {
        log.info("数据表注册：{} - {}", databaseName, tableName);
        this.listenerMap.put(genKey(databaseName, tableName), listener);
    }

    @Override
    public void onEvent(Event event) {
        EventType type = event.getHeader().getEventType();
        log.info("event type: {}", type);
        if (type == EventType.TABLE_MAP) {
            TableMapEventData data = event.getData();
            this.databaseName = data.getDatabase();
            this.tableName = data.getTable();
        }

        if (type != EventType.EXT_UPDATE_ROWS
                && type != EventType.EXT_WRITE_ROWS
                && type != EventType.EXT_DELETE_ROWS) {
            return;
        }

        if (StringUtils.isEmpty(databaseName) || StringUtils.isEmpty(tableName)) {
            return;
        }

        String key = genKey(this.databaseName, this.tableName);
        Listener listener = this.listenerMap.get(key);
        if (null == listener) {
            log.info("跳过监听器,key: {}", key);
            return;
        }
        log.info("触发事件: {}", type.name());
        try {
            BinlogRowData rowData = buildRowData(event.getData());
            if (rowData == null) {
                return;
            }
            rowData.setEventType(type);
            listener.onEvent(rowData);
        } catch (Exception ex) {
            ex.printStackTrace();
            log.error(ex.getLocalizedMessage());
        } finally {
            this.tableName = "";
            this.databaseName = "";
        }
    }

    private List<Serializable[]> getAfterValues(EventData eventData) {
        if (eventData instanceof WriteRowsEventData) {
            return ((WriteRowsEventData) eventData).getRows();
        }

        if (eventData instanceof UpdateRowsEventData) {
            return ((UpdateRowsEventData) eventData).getRows().stream()
                    .map(Map.Entry::getValue)
                    .collect(Collectors.toList());
        }

        if (eventData instanceof DeleteRowsEventData) {
            return ((DeleteRowsEventData) eventData).getRows();
        }

        return Collections.emptyList();
    }
    private BinlogRowData buildRowData(EventData eventData) {
        TableTemplate table = templateHolder.getTable(tableName);
        if (null == table) {
            log.warn("无法找到表{}的数据", tableName);
            return null;
        }
        List<Map<String, String>> afterMapList = new ArrayList<>();
        for (Serializable[] afterValue : getAfterValues(eventData)) {
            Map<String, String> afterMap = new HashMap<>();
            int collen = afterValue.length;

            for (int x = 0; x < collen; x++) {
                // 取出当前索引对应的列名
                String colName = table.getPosMap().get(x);
                if (null == colName) {
                    continue;
                }
                String colValue = afterValue[x].toString();
                afterMap.put(colName, colValue);
            }
            afterMapList.add(afterMap);
        }
        BinlogRowData rowData = new BinlogRowData();
        rowData.setAfter(afterMapList);
        rowData.setTable(table);

        return rowData;
    }
}
